package kino.aof;

import io.netty.channel.Channel;
import kino.client.KinoClient;
import kino.common.KinoConstant;
import kino.server.KinoServer;
import org.apache.commons.io.FileUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

@Component
public class AofService implements InitializingBean {
    /**
     * aof文件记录执行器 用于对aof文件的异步写入
     */
    private ExecutorService aofExecutor;
    /**
     * aof文件
     */
    private File aofFile;

    /**
     * aof文件地址
     */
    @Value("${kino.aof.file.path}")
    private String aofFilePath;

    public void aofAppend(KinoClient client) {
        // 获取客户端当前的数据库编码
        String dbNo = String.valueOf(client.getDb().getDbNo());
        // aof需要记录的原始命令
        byte[] oriQueryBuf = client.getQuerybuf();
        // 写入aof的动作异步执行，原始命令和当前客户端编号在上面已获取到栈内不用担心被后面的客户端覆盖，后面需要处理的命令可以先去执行了，此处异步去写文件就好了
        aofExecutor.execute(() -> {
            try {
                // 数据库切换
                String selectCommand = "*2"+ KinoConstant.crlf +"$6" + KinoConstant.crlf + "SELECT" + KinoConstant.crlf + "$" + dbNo.length() + KinoConstant.crlf + dbNo + KinoConstant.crlf;
                // 先记录数据切换命令
                FileUtils.writeByteArrayToFile(aofFile, selectCommand.getBytes(), true);
                // 再记录原始命令
                FileUtils.writeByteArrayToFile(aofFile, oriQueryBuf, true);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }

    /**
     * kino server启动 从磁盘中加载aof并执行内容使db状态恢复到关闭前
     * @param kinoServer
     */
    public void loadDataFromDisk(KinoServer kinoServer) {
        // 伪客户端
        KinoClient fakeClient = new KinoClient(kinoServer.defaultDb(), (Channel) null, kinoServer.getCommandInterpreter());
        fakeClient.setAofClient(true);
        kinoServer.addClient(fakeClient);
        List<String> newCommandList = new ArrayList<>();
        StringBuffer command = new StringBuffer();
        try {
            List<String> commandList = FileUtils.readLines(aofFile, Charset.defaultCharset());
            commandList = commandList.stream().filter(item -> !item.startsWith("$")).collect(Collectors.toList());
            commandList.forEach(item -> {
                if (item.startsWith("*")) {
                    if (command.length() > 0) {
                        newCommandList.add(command.toString());
                    }
                    command.setLength(0);
                } else {
                    if (command.length() != 0) {
                        command.append(" ");
                    }
                    command.append(item);
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (command.length() != 0) {
                newCommandList.add(command.toString());
            }
        }
        // 执行aof命令
        for (String clintQuery : newCommandList) {
            fakeClient.setQueryBuffer(new StringBuffer(clintQuery));
            fakeClient.execute();
        }
        // 卸载aof客户端
        kinoServer.clientLoginOut(fakeClient);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        // 创建线aof追加线程池
        this.aofExecutor = Executors.newSingleThreadExecutor();
        // 初始化aof文件
        this.aofFile = new File(aofFilePath);
    }
}
